package com.galeno;

import akka.stream.impl.io.FileSource;
import com.galeno.day02.NoParallSourceFunc;
import com.galeno.day02.NoParallSourceFunc2;
import com.galeno.day02.ParallSourceFun;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * @author galeno
 * @Title:
 * @Description:封装Flink常用的 Source 和 Sink
 * @date 2021/10/1522:41
 */
public class FlinkUtils {
    public static String LinuxURL = "192.168.77.101";
    public static int port = 8888;
    public static String kafkaBootstrapAddr="galeno01:9092,galeno02:9092,galeno03:9092";
    //public static String kafkaBootstrapAddr="galeno01:2181,galeno02:9092,galeno03:9092";
    public static String Linux3="galeno01";
    public static StreamExecutionEnvironment env;

    public static StreamExecutionEnvironment getEnv() {
        return StreamExecutionEnvironment.getExecutionEnvironment();
    }

    public static StreamExecutionEnvironment getLocEnvWebUI(Configuration conf) {
        env= StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        return env;
    }

    public void run() throws Exception {
        env.execute();
    }


    public static NoParallSourceFunc getNoParSourceFunc(){
        return new NoParallSourceFunc();
    }
    public static NoParallSourceFunc2 getNoParSourceFunc2(){
        return new NoParallSourceFunc2();
    }
    public static ParallSourceFun getParallSourceFun(){
        return new ParallSourceFun();
    }

    /**
     * 读完即停
     * @param filePath 传入的路径
     * @param conf  传入参数
     * @return
     */
    public static DataStreamSource<String> getFileSourceEnv(String filePath, Configuration conf){
        StreamExecutionEnvironment locEnvWebUI = getLocEnvWebUI(conf);
        return locEnvWebUI.readTextFile(filePath);
    }
    /**
     *读完监听,程序不停止
     */
    public static DataStreamSource<String>  getFileSourceNotToStop(String filePath,Configuration conf){
        StreamExecutionEnvironment locEnvWebUI = getLocEnvWebUI(conf);
        return locEnvWebUI.readFile(new TextInputFormat(new Path(filePath)),
                filePath, FileProcessingMode.PROCESS_CONTINUOUSLY,1000);
    }

    /**
     *
     * @param conf 配置
     * @param bingxingdu 并行度
     * @param groupid   组id
     * @param kafkaoffset  kafka出现问题,偏移量位置   earlist /latest
     * @param consumerTopic 消费主题
     * @return
     */
    public static FlinkKafkaConsumer<String> getKafkaSource(Configuration conf,int bingxingdu,String groupid,String kafkaoffset,String consumerTopic){
        StreamExecutionEnvironment locEnvWebUI = getLocEnvWebUI(conf);
        locEnvWebUI.setParallelism(bingxingdu);
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","galeno01:9092,galeno02:9092,galeno03:9092");
        properties.setProperty("group.id",groupid);
        properties.setProperty("auto.offset.rest",kafkaoffset);
        return new FlinkKafkaConsumer<String>(consumerTopic,new SimpleStringSchema(),properties);


    }












}
